package org.redis.cluster.command;

import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashSet;

import org.apache.commons.lang3.ArrayUtils;
import org.redis.cluster.Ordered;
import org.redis.cluster.RedisCluster;
import org.redis.cluster.interceptor.Interceptor;
import org.redis.cluster.interceptor.Invocation;
import org.redis.cluster.plugin.Plugin;
import org.redis.cluster.redis.RedisCommand;
import org.redis.cluster.redis.transport.RedisReply;
import org.redis.cluster.redis.transport.reply.RedisErrorReply;
import org.redis.cluster.transport.session.Session;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class UnsupportedCommandProcessor implements Plugin,Interceptor,Ordered {
	
	private final static String COMMAND_SPLITOR=",";
	
	private final static String UNSUPPOTED_COMMAND_CONFIG="unsupported_commands";
	
	private final static String DEFAULT_UNSUPPORTED_COMMANDS="DUMP,MIGRATE,MOVE,RANDOMKEY,SORT,BLPOP,BRPOP,BRPOPLPUSH,KEYS,PSUBSCRIBE,PUBLISH,PUBSUB,PUNSUBSCRIBE"
			+ ",SUBSCRIBE,UNSUBSCRIBE,DISCARD,EXEC,MULTI,UNWATCH,WATCH,EVAL,EVALSHA,SCRIPT EXISTS,SCRIPT FLUSH,SCRIPT KILL,SCRIPT LOAD,SELECT"
			+ ",BGREWRITEAOF,BGSAVE,CLIENT GETNAME,CLIENT KILL,CLIENT LIST,CLIENT SETNAME,CONFIG GET,CONFIG RESETSTAT,CONFIG REWRITE,DBSIZE,DEBUG OBJECT"
			+ ",DEBUG SEGFAULT,FLUSHALL,FLUSHDB,LASTSAVE,MONITOR,PSYNC,SAVE,SHUTDOWN,SLAVEOF,SLOWLOG,SYNC,TIME";
	
	protected final Logger logger=LoggerFactory.getLogger(getClass());
	
	private RedisCluster redisCluster;
	
	private Collection<String> unsupportedCommands;
	
	@Override
	public void init(RedisCluster redisCluster) {
		this.redisCluster=redisCluster;
		this.unsupportedCommands=initUnsupportedCommands();
	}
	
	@Override
	public void destroy() {
	}
	
	protected Collection<String> initUnsupportedCommands() {
		String unsupportedCommandStr=redisCluster.getConfigManager().getSystemConfig().get(UNSUPPOTED_COMMAND_CONFIG).getValue(String.class,DEFAULT_UNSUPPORTED_COMMANDS);
		String splitedUnsupportedCommandStr[]=unsupportedCommandStr.split(COMMAND_SPLITOR);
		if(ArrayUtils.isEmpty(splitedUnsupportedCommandStr)) {
			return Collections.emptySet();
		}
		
		return new HashSet<String>(Arrays.asList(splitedUnsupportedCommandStr));
	}

	@Override
	public int getOrder() {
		return Ordered.HIGH;
	}

	@Override
	public RedisReply intercept(Invocation invocation) throws Throwable {
		RedisCommand command=invocation.getCommand();
		if(isSupport(command)) {
			return invocation.invoke();
		}
		
		return doUnsupportProcess(command, invocation.getSession());
	}
	
	protected boolean isSupport(RedisCommand command) {
		return !unsupportedCommands.contains(command.getCommand());
	}
	
	protected RedisReply doUnsupportProcess(RedisCommand command,Session session) {
		if(logger.isWarnEnabled()) {
			logger.warn("不支持的命令 - {}",command.getCommand());
		}
	
		return new RedisErrorReply("unsupported command - "+command.getCommand());
	}
}